package DianShang_2024.ds_server.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, date_format, lit, to_timestamp}

object extract04 {
  def main(args: Array[String]): Unit = {
    /*
            4、将ods库中order_detail表昨天的分区（任务一生成的分区）数据抽取到dwd库中fact_order_detail的动态分区表，分区字段为etl_date，类型为String，
            取create_time值并将格式转换为yyyyMMdd，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列，
            其中dwd_insert_user、dwd_modify_user均填写“user1”，dwd_insert_time、dwd_modify_time均填写当前操作时间，并进行数据类型转换。
            使用hive cli执行show partitions dwd.fact_order_detail命令；
    提示：若完成实时任务二中的第3小题后，可将hbase中的数据合并后在进行清洗，清洗后数据添加至DWD中
     */

    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第四题")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd_server")
    //   设置动态分区
    spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")

    //      因为一些原因，具体不清楚，反正不使用string数据类型会报错
    //  去虚拟机hive的shell操作执行: alter table fact_order_master change column dwd_insert_time dwd_insert_time string;
    //                            alter table fact_order_master change column dwd_modify_time dwd_modify_time string;


    spark.sql(
      """
        |select
        |order_detail_id ,
        |order_sn ,
        |product_id ,
        |product_name ,
        |product_cnt ,
        |product_price ,
        |average_cost ,
        |weight ,
        |fee_money ,
        |w_id ,
        |substr(create_time,1,8) as create_time,
        |modified_time ,
        |etl_date
        |from ods_server.order_detail01
        |where etl_date='20240311'
        |""".stripMargin)
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss")
      )
      .withColumn("dwd_modify_user",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss")
      )
      .createOrReplaceTempView("temp_table")

spark.sql("select create_time from temp_table limit 10").show

    //  抽取数据
    spark.sql(
      """
        |insert into table dwd_server.fact_order_detail
        |partition(etl_date)
        |select
        |*
        |from temp_table
        |""".stripMargin)












    //  关闭 spaksql的环境
    spark.close()


  }

}
